/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package homomorphism.list;

import homomorphism.HomomorphismBase;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.ArrayList;
import java.util.Iterator;
import java.io.IOException;
import java.lang.reflect.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import utility.LongPair;

/**
 *
 * @author takeshi
 */
public abstract class SecondReducerFor3rdHomo<InVal, OtVal> extends MapReduceBase implements Reducer<LongPair, InVal, NullWritable, OtVal> {

    private static final Log LOG = LogFactory.getLog(SecondReducerFor3rdHomo.class);
    private static final String userAppProperty = HomomorphismBase.UserClassNameProperty;
 //   private Class FoldingType;
    private Class UnFoldingType;
    private String UserClassName;

    @Override
    public void configure(JobConf job) {
        super.configure(job);
        UserClassName = job.get(userAppProperty);
  //      FoldingType = job.getMapOutputValueClass();
        UnFoldingType = job.getOutputValueClass();
    }

    public OtVal fold(ArrayList<InVal> values) throws Exception {
        ClassLoader myloader = FirstReducerFor3rdHomo.class.getClassLoader();
        Class UserApp;
        if (UserClassName == null || UserClassName.isEmpty()) {
            return null;
        }
        try {
            UserApp = myloader.loadClass(UserClassName);
            Class partypes[] = new Class[1];
            ArrayList<? extends Writable> eee = new ArrayList();
            partypes[0] = values.getClass();
            Method _fold = UserApp.getMethod("fold", partypes);

            Object arglist[] = new Object[1];
            arglist[0] = values;
            Constructor ct = UserApp.getConstructor(); //partypes
            Object methobj = ct.newInstance();
            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
            return rst;
        } catch (ClassNotFoundException ee) {
        }
        return null;

    }
    //new version
//    public OtVal fold(ListIterator<InVal> values) throws Exception {
//        if (UserClassName == null || UserClassName.isEmpty()) {
//            throw new IOException("2nd phase: null fold:UserClassName\n");
//        }
//        try {
//            Class UserApp = Class.forName(UserClassName);
//            Class partypes[] = new Class[1];
//            ArrayList<? extends Writable> eee = new ArrayList();
//            partypes[0] = values.getClass();
//            Method _fold = UserApp.getMethod("fold", partypes);
//            Object arglist[] = new Object[1];
//            arglist[0] = values;
//            Constructor ct = UserApp.getConstructor(); //partypes
//            Object methobj = ct.newInstance();
//            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
//            return rst;
//
//        } catch (ClassNotFoundException ee) {
//            LOG.error(ee.getMessage());
//        }
//        return null;
//    }
    //Dynamic Class Loading and Reloading in Java
    public ArrayList<InVal> unfold(OtVal value) throws Exception {
        Class UserApp = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = UnFoldingType;
        Method _unfold = UserApp.getMethod("unfold", partypes);

        Object arglist[] = new Object[1];
        arglist[0] = value;
        Constructor ct = UserApp.getConstructor();//partypes
        Object methobj = ct.newInstance();
        ArrayList<InVal> rst = (ArrayList<InVal>) _unfold.invoke(methobj, arglist);
        return rst;
    }

    @Override
    public void reduce(LongPair key, Iterator<InVal> values, OutputCollector<NullWritable, OtVal> output, Reporter reporter)
            throws IOException {

        OtVal outVal = null;
        ArrayList<InVal> tmp1 = new ArrayList<InVal>(1);
        ArrayList<InVal> tmp2 = new ArrayList<InVal>();

        int count = 0;
        while (values.hasNext()) {
            try {
                InVal pr = values.next();
                tmp1.add(pr);
                LOG.info("2nd  reduce : the " + count + " input element  " + pr + ",  key  is " + key);
                if (outVal == null) {// for the first element
                    outVal = fold(tmp1);
                    tmp1.clear();
                } else { // for the rest
                    tmp2.clear();
                    for (InVal e : unfold(outVal)) {
                        if (e == null) {
                            throw new IOException("2.2 reduce null exception : outValPair\n");
                        }
                        tmp2.add(e);
                    }
                    tmp2.add(tmp1.get(0));
                    outVal = fold(tmp2);
                    tmp1.clear();
                }
            } catch (Exception ex) {
                Logger.getLogger(FirstReducerFor3rdHomo.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        if (outVal == null) {
            throw new IOException("null exception : outValPair\n");
        }
        output.collect(NullWritable.get(), outVal);
    }
}
